use std::{
    collections::VecDeque,
    sync::{
        atomic::{AtomicU64, AtomicU8},
        Arc,
    },
};

use parking_lot::Mutex;
use tokio::{
    io::AsyncWriteExt,
    net::tcp::{OwnedReadHalf, OwnedWriteHalf},
    sync::{oneshot, Mutex as TokioMutex},
};

use crate::{
    error::RedisError,
    resp::{Command, FrameBuffer, Value},
    RedisResult,
};

use super::{
    shutdown::{ShutdownReason, ShutdownSignal},
    ConnectionState, ConnectionStatus,
};

pub struct NodeSocket {
    pub(crate) host: String,
    pub(crate) port: u32,
    write_half: TokioMutex<Option<OwnedWriteHalf>>,
    status: AtomicU8,
    epoch: AtomicU64,
    pending: Mutex<VecDeque<oneshot::Sender<Value>>>,
}

impl NodeSocket {
    pub fn new(host: String, port: u32) -> Self {
        Self {
            host,
            port,
            write_half: TokioMutex::new(None),
            status: AtomicU8::new(ConnectionStatus::Initialize as u8),
            epoch: AtomicU64::new(0),
            pending: Mutex::new(VecDeque::new()),
        }
    }

    pub async fn send(&self, content: &[u8]) -> RedisResult<oneshot::Receiver<Value>> {
        // log::trace!("Send {content:?}");
        let (tx, rx) = oneshot::channel::<Value>();
        let mut guard = self.write_half.lock().await;
        if let Some(write_half) = guard.as_mut() {
            self.pending_push_front(tx);
            match write_half.write_all(content).await {
                Ok(_) => Ok(rx),
                Err(err) => {
                    self.pending_pop_front();
                    Err(RedisError::Send(format!("{:?}", err)))
                }
            }
        } else {
            Err(RedisError::NoStream)
        }
    }

    pub(crate) async fn send_no_reply(&self, content: &[u8]) -> RedisResult<()> {
        let mut guard = self.write_half.lock().await;
        if let Some(write_half) = guard.as_mut() {
            match write_half.write_all(content).await {
                Ok(_) => Ok(()),
                Err(err) => Err(RedisError::Send(format!("{:?}", err))),
            }
        } else {
            Err(RedisError::NoStream)
        }
    }

    pub fn set_connecting(&self) {
        self.status.store(
            ConnectionStatus::Connecting as u8,
            std::sync::atomic::Ordering::Release,
        );
    }

    pub fn set_reconnecting(&self) {
        self.status.store(
            ConnectionStatus::Reconnecting as u8,
            std::sync::atomic::Ordering::Release,
        );
    }

    pub async fn set_disconnected(&self) {
        self.status.store(
            ConnectionStatus::Disconnected as u8,
            std::sync::atomic::Ordering::Release,
        );
        let mut guard = self.write_half.lock().await;
        if let Some(mut write_half) = guard.take() {
            _ = write_half.shutdown();
        }
        self.pending_clear();
    }

    pub async fn set_connected(&self, write_half: OwnedWriteHalf) {
        self.status.store(
            ConnectionStatus::Disconnected as u8,
            std::sync::atomic::Ordering::Release,
        );
        let mut guard = self.write_half.lock().await;
        if let Some(mut original) = guard.replace(write_half) {
            _ = original.shutdown();
        }
    }

    // pub fn get_connection_status(&self) -> ConnectionStatus {
    //     self.status.store(
    //         ConnectionStatus::Disconnected as u8,
    //         std::sync::atomic::Ordering::Release,
    //     );
    //     let val = self.status.load(std::sync::atomic::Ordering::Acquire);
    //     ConnectionStatus::from(val)
    // }

    /// return previous epoch
    pub(crate) fn epoch_advance(&self) -> u64 {
        self.epoch.fetch_add(1, std::sync::atomic::Ordering::AcqRel)
    }

    /// return new value
    pub(crate) fn epoch_compare_advance(&self, cur: u64) -> bool {
        let new = if cur == u64::MAX { 0 } else { cur + 1 };
        match self.epoch.compare_exchange(
            cur,
            new,
            std::sync::atomic::Ordering::Acquire,
            std::sync::atomic::Ordering::Relaxed,
        ) {
            Ok(_) => true,
            Err(_) => false,
        }
    }

    /// get current epoch
    pub(crate) fn epcoh_get(&self) -> u64 {
        self.epoch.load(std::sync::atomic::Ordering::Acquire)
    }

    fn pending_push_front(&self, value: oneshot::Sender<Value>) {
        let mut guard = self.pending.lock();
        guard.push_front(value);
    }

    fn pending_pop_front(&self) {
        let mut guard = self.pending.lock();
        _ = guard.pop_front();
    }

    fn pending_pop_back(&self) -> Option<oneshot::Sender<Value>> {
        let mut guard = self.pending.lock();
        guard.pop_back()
    }

    fn pending_clear(&self) {
        let mut guard = self.pending.lock();
        guard.clear();
    }
}

pub(super) fn start_stream_read(
    state: Arc<ConnectionState>,
    socket: Arc<NodeSocket>,
    read_half: OwnedReadHalf,
    shutdown_sender: tokio::sync::mpsc::Sender<ShutdownSignal>,
) {
    let epoch = socket.epcoh_get();
    if let Some(timeout) = state.timeout {
        let secs = timeout.as_secs();
        if secs > 5 {
            tokio::spawn(start_stream_read_internal_timeout(
                read_half,
                shutdown_sender,
                epoch,
                state,
                socket,
                timeout,
            ));

            return;
        }
    }
    tokio::spawn(async move {
        start_stream_read_internal(&read_half, &shutdown_sender, epoch, &state, &socket).await;
    });
}

async fn start_stream_read_internal_timeout(
    read_half: OwnedReadHalf,
    shutdown_sender: tokio::sync::mpsc::Sender<ShutdownSignal>,
    epoch: u64,
    state: Arc<ConnectionState>,
    socket: Arc<NodeSocket>,
    timeout: std::time::Duration,
) {
    let cmd_ping = &Command::new("PING").arg("1").encode();
    loop {
        tokio::select! {
            _ =  tokio::time::sleep(timeout) => {
                if !state.stream_flag.load(std::sync::atomic::Ordering::Relaxed) {
                    let _ = socket.send(cmd_ping).await;
                }
            }
            _ = start_stream_read_internal(&read_half, &shutdown_sender, epoch,&state,&socket) => {
                return;
            }
        }
    }
}

async fn start_stream_read_internal(
    read_half: &OwnedReadHalf,
    shutdown_sender: &tokio::sync::mpsc::Sender<ShutdownSignal>,
    epoch: u64,
    state: &Arc<ConnectionState>,
    socket: &Arc<NodeSocket>,
) {
    let mut frame_buffer = FrameBuffer::new();
    let mut buf = vec![0; 2048];
    let mut retry = 0u8;
    loop {
        match read_half.ready(tokio::io::Interest::READABLE).await {
            Ok(ready) => {
                if ready.is_read_closed() || ready.is_write_closed() {
                    _ = shutdown_sender
                        .send(ShutdownSignal::new(
                            ShutdownReason::ServerClosed1,
                            epoch,
                            state.clone(),
                            socket.clone(),
                            None,
                        ))
                        .await;
                    return;
                }
                if !ready.is_readable() {
                    continue;
                }
                match read_half.try_read(&mut buf) {
                    Ok(0) => {
                        if retry == 30 {
                            _ = shutdown_sender
                                .send(ShutdownSignal::new(
                                    ShutdownReason::ServerClosed2,
                                    epoch,
                                    state.clone(),
                                    socket.clone(),
                                    None,
                                ))
                                .await;
                            return;
                        }
                        retry += 1;
                    }
                    Ok(n) => {
                        frame_buffer.put(&buf[0..n]);
                        match frame_buffer.read_next_values() {
                            Ok(vals) => {
                                if vals.len() > 0 {
                                    if state.stream_flag.load(std::sync::atomic::Ordering::Relaxed)
                                    {
                                        let guard = state.stream_tx.lock().await;
                                        if let Some(tx) = guard.as_ref() {
                                            for val in vals {
                                                match tx.send(val).await {
                                                    Ok(_) => {}
                                                    Err(_) => {
                                                        log::error!(
                                                            "error on send stream response value"
                                                        );
                                                    }
                                                }
                                            }
                                        }
                                    } else {
                                        for val in vals {
                                            if let Some(tx) = socket.pending_pop_back() {
                                                match tx.send(val) {
                                                    Ok(_) => {}
                                                    Err(_) => {
                                                        // log::error!("error on send response value, {val:?}");
                                                    }
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                            Err(err) => {
                                _ = shutdown_sender
                                    .send(ShutdownSignal::new(
                                        ShutdownReason::ServerClosed3,
                                        epoch,
                                        state.clone(),
                                        socket.clone(),
                                        Some(err),
                                    ))
                                    .await;
                                break;
                            }
                        }
                    }
                    Err(ref err) if err.kind() == std::io::ErrorKind::WouldBlock => {
                        continue;
                    }
                    Err(err) => {
                        log::error!("{}", err);
                        return;
                    }
                }
            }
            Err(err) => {
                _ = shutdown_sender
                    .send(ShutdownSignal::new(
                        ShutdownReason::ServerClosed4,
                        epoch,
                        state.clone(),
                        socket.clone(),
                        Some(RedisError::new_socket_disconnected(
                            "socket disconnected",
                            err,
                        )),
                    ))
                    .await;
                break;
            }
        }
    }
}
